package service.pqueue;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public class PHashQueue implements IPQueue {
	private static Object lock = new Object();
	private long polls = 0;
	private long offers = 0;
	private ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
	private ConcurrentHashMap<String, String> map = new ConcurrentHashMap<String, String>();

	public boolean offer(String key, String msg) {
		synchronized (lock) {
			try {
				if (map.containsKey(key)) {
					map.remove(key);
					map.put(key, msg);
				} else {
					map.put(key, msg);
					queue.offer(key);
				}
			} catch (Exception ex) {

			}
		}

		incOffers();

		return true;
	}

	private void incOffers() {
		if (offers >= Long.MAX_VALUE) {
			offers = 0;
		}
		offers++;
	}

	public String poll() {
		incPolls();

		synchronized (lock) {
			try {
				String key = queue.poll();
				String ret = null;
				if (map.containsKey(key)) {
					ret = map.get(key);
					map.remove(key);
				}

				return ret;
			} catch (Exception ex) {
			}
			return null;
		}
	}

	private void incPolls() {
		if (polls >= Long.MAX_VALUE) {
			polls = 0;
		}
		polls++;
	}

	public int statsLen() {
		return queue.size();
	}

	public long statsOffers() {
		return offers;
	}

	public long statsPolls() {
		return polls;
	}

	public void flush() {
		synchronized (lock) {
			try {
				queue.clear();
				map.clear();
			} catch (Exception ex) {
			}
		}
	}
}
